# consumer.py
import json
from kafka import KafkaConsumer
import redis

# Kafka 配置
bootstrap_servers = ['kafka1:9092', 'kafka2:9093', 'kafka3:9094']
topic = 'user-behavior'
group_id = 'behavior-analytics-group'

# Redis 配置
redis_client = redis.Redis(host='redis', port=6379, db=0, decode_responses=True)

# 创建消费者实例
consumer = KafkaConsumer(
    topic,
    bootstrap_servers=bootstrap_servers,
    group_id=group_id,
    auto_offset_reset='earliest',
    enable_auto_commit=False,  # 关闭自动提交
    max_poll_records=100,      # 每次拉取的最大消息数
    max_poll_interval_ms=300000  # 处理消息的最大间隔
)

def process_message(message):
    print(f"消费到消息: {message}")  # 打印消费到的数据

    # 1. 基于业务 Key 去重（如订单 ID、用户操作）
    business_key = f"{message['user_id']}_{message['action']}_{message['timestamp']}"
    
    # 2. 检查是否已处理过（可使用 Redis 或数据库）
    if redis_client.exists(business_key):
        print(f"重复消息，跳过处理: {business_key}")
        return
    
    # 3. 处理业务逻辑
    try:
        # 数据库操作...
        # db.execute("INSERT INTO user_actions (...) VALUES (...)")
        
        # 4. 处理成功后记录已处理
        redis_client.setex(business_key, 86400, "processed")  # 缓存 24 小时
    except Exception as e:
        print(f"处理失败: {e}")
        raise  # 抛出异常，不提交 Offset

if __name__ == "__main__":
    print("Kafka 消费者已启动，开始消费用户行为数据...")

    # 消费消息
    try:
        for message in consumer:
            process_message(json.loads(message.value))  # 反序列化消息
            consumer.commit()  # 手动提交当前 Offset
    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()